package com.central.scheduler.master.registry;

import com.central.common.model.WorkerServer;
import com.central.scheduler.service.queue.MasterPriorityQueue;
import com.central.scheduler.common.enums.NodeType;
import com.central.scheduler.common.utils.NetUtils;
import com.central.scheduler.dao.AlertDao;
import com.central.scheduler.service.registry.RegistryClient;
import com.central.scheduler.dao.entity.Server;
import com.central.scheduler.dao.mapper.WorkerGroupMapper;
import com.central.scheduler.dao.model.WorkerGroup;
import com.central.scheduler.remote.utils.NamedThreadFactory;
import com.central.scheduler.spi.register.DataChangeEvent;
import com.central.scheduler.spi.register.SubscribeListener;
import org.apache.commons.collections.CollectionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

import static com.central.scheduler.common.Constants.REGISTRY_SCHEDULER_MASTERS;
import static com.central.scheduler.common.Constants.REGISTRY_SCHEDULER_WORKERS;

/**
 * @author Tindy
 * @date 2021/11/4
 * @describe
 */
@Service
public class ServerNodeManager implements InitializingBean {

    private final Logger logger = LoggerFactory.getLogger(ServerNodeManager.class);

    /**
     * master lock
     */
    private final Lock masterLock = new ReentrantLock();

    /**
     * worker group lock
     */
    private final Lock workerGroupLock = new ReentrantLock();

    /**
     * worker node info lock
     */
    private final Lock workerNodeInfoLock = new ReentrantLock();

    /**
     * worker group nodes
     */
    private final ConcurrentHashMap<Long, Set<String>> workerGroupNodes = new ConcurrentHashMap<>();

    /**
     * master nodes
     */
    private final Set<String> masterNodes = new HashSet<>();

    /**
     * worker node info
     */
    private final Map<String, String> workerNodeInfo = new HashMap<>();

    /**
     * executor service
     */
    private ScheduledExecutorService executorService;

    /**
     * zk client
     */
    private RegistryClient registryClient = RegistryClient.getInstance();

    /**
     * eg : /node/worker/group/127.0.0.1:xxx
     */
    private static final int WORKER_LISTENER_CHECK_LENGTH = 5;
    private MasterPriorityQueue masterPriorityQueue = new MasterPriorityQueue();
    /**
     * worker group mapper
     */
    @Autowired
    private WorkerGroupMapper workerGroupMapper;
    /**
     * alert dao
     */
    @Autowired
    private AlertDao alertDao;

    public static volatile List<Integer> SLOT_LIST = new ArrayList<>();

    public static volatile Integer MASTER_SIZE = 0;
    public static Integer getSlot() {
        if (SLOT_LIST.size() > 0) {
            return SLOT_LIST.get(0);
        }
        return 0;
    }
    @Override
    public void afterPropertiesSet() throws Exception {
        /**
         * load nodes from zookeeper
         */
        load();
        /**
         * init executor service
         */
        executorService = Executors.newSingleThreadScheduledExecutor(new NamedThreadFactory("ServerNodeManagerExecutor"));
        executorService.scheduleWithFixedDelay(new WorkerNodeInfoAndGroupDbSyncTask(), 0, 10, TimeUnit.SECONDS);
        /**
         * init MasterNodeListener listener
         */
        registryClient.subscribe(REGISTRY_SCHEDULER_MASTERS, new MasterDataListener());
        /**
         * init WorkerNodeListener listener
         */
        registryClient.subscribe(REGISTRY_SCHEDULER_WORKERS, new WorkerDataListener());
    }
    /**
     * load nodes from zookeeper
     */
    public void load() {
        /**
         * master nodes from zookeeper
         */
        updateMasterNodes();
    }
    /**
     * sync worker group nodes
     *
     * @param workerGroupId workerGroupId
     * @param nodes worker nodes
     */
    private void syncWorkerGroupNodes(Long workerGroupId, Set<String> nodes) {
        workerGroupLock.lock();
        try {
            Set<String> workerNodes = workerGroupNodes.getOrDefault(workerGroupId, new HashSet<>());
            workerNodes.clear();
            workerNodes.addAll(nodes);
            workerGroupNodes.put(workerGroupId, workerNodes);
        } finally {
            workerGroupLock.unlock();
        }
    }
    private void updateMasterNodes() {
        SLOT_LIST.clear();
        this.masterNodes.clear();
        String nodeLock = registryClient.getMasterLockPath();
        try {
            registryClient.getLock(nodeLock);
            Set<String> currentNodes = registryClient.getMasterNodesDirectly();
            List<Server> masterNodes = registryClient.getServerList(NodeType.MASTER);
            syncMasterNodes(currentNodes, masterNodes);
        } catch (Exception e) {
            logger.error("update master nodes error", e);
        } finally {
            registryClient.releaseLock(nodeLock);
        }
    }

    /**
     * sync master nodes
     *
     * @param nodes master nodes
     */
    private void syncMasterNodes(Set<String> nodes, List<Server> masterNodes) {
        masterLock.lock();
        try {
            this.masterNodes.addAll(nodes);
            this.masterPriorityQueue.clear();
            this.masterPriorityQueue.putList(masterNodes);
            int index = masterPriorityQueue.getIndex(NetUtils.getHost());
            if (index >= 0) {
                MASTER_SIZE = nodes.size();
                SLOT_LIST.add(masterPriorityQueue.getIndex(NetUtils.getHost()));
            }
            logger.info("update master nodes, master size: {}, slot: {}",
                    MASTER_SIZE, SLOT_LIST.toString()
            );
        } finally {
            masterLock.unlock();
        }
    }
    /**
     * get worker group nodes
     *
     * @param workerGroupId workerGroupId
     * @return worker nodes
     */
    public Set<String> getWorkerGroupNodes(Long workerGroupId) {
        workerGroupLock.lock();
        try {
            Set<String> nodes = workerGroupNodes.get(workerGroupId);
            if (CollectionUtils.isNotEmpty(nodes)) {
                return Collections.unmodifiableSet(nodes);
            }
            return nodes;
        } finally {
            workerGroupLock.unlock();
        }
    }

    public Map<Long, Set<String>> getWorkerGroupNodes() {
        return Collections.unmodifiableMap(workerGroupNodes);
    }
    /**
     * sync worker node info
     *
     * @param newWorkerNodeInfo new worker node info
     */
    private void syncWorkerNodeInfo(Map<String, String> newWorkerNodeInfo) {
        workerNodeInfoLock.lock();
        try {
            workerNodeInfo.clear();
            workerNodeInfo.putAll(newWorkerNodeInfo);
        } finally {
            workerNodeInfoLock.unlock();
        }
    }
    /**
     * worker node info and worker group db sync task
     */
    class WorkerNodeInfoAndGroupDbSyncTask implements Runnable {

        @Override
        public void run() {
            updateWorkerNodes();
        }
    }

    private void updateWorkerNodes() {
        // sync worker node info
        Map<String, String> newWorkerNodeInfo = registryClient.getServerMaps(NodeType.WORKER);
        syncWorkerNodeInfo(newWorkerNodeInfo);

        // sync worker group nodes from database
        List<WorkerGroup> workerGroupList = workerGroupMapper.selectList(null);
        if (CollectionUtils.isNotEmpty(workerGroupList)) {
            for (WorkerGroup wg : workerGroupList) {
                Long workerGroupId = wg.getId();
                Set<String> nodes = new HashSet<>();
                List<WorkerServer> workerServers = wg.getWorkerServers();
                for (WorkerServer workerServer : workerServers) {
                    if (newWorkerNodeInfo.containsKey(workerServer.getHost())) {
                        nodes.add(workerServer.getHost());
                    }
                }
                if (!nodes.isEmpty()) {
                    syncWorkerGroupNodes(workerGroupId, nodes);
                }
            }
        }
    }

    /**
     * worker group node listener
     */
    class WorkerDataListener implements SubscribeListener {

        @Override
        public void notify(String path, DataChangeEvent dataChangeEvent) {
            if (registryClient.isWorkerPath(path)) {
                try {
                    if (dataChangeEvent == DataChangeEvent.ADD) {
                        logger.info("worker group node : {} added.", path);
                        updateWorkerNodes();
                    } else if (dataChangeEvent == DataChangeEvent.REMOVE) {
                        logger.info("worker group node : {} down.", path);
                        updateWorkerNodes();
                        alertDao.sendServerStopedAlert(1, path, "WORKER");
                    }
                } catch (IllegalArgumentException ex) {
                    logger.warn(ex.getMessage());
                } catch (Exception ex) {
                    logger.error("WorkerGroupListener capture data change and get data failed", ex);
                }

            }
        }
    }

    /**
     * master node listener
     */
    class MasterDataListener implements SubscribeListener {
        @Override
        public void notify(String path, DataChangeEvent dataChangeEvent) {
            if (registryClient.isMasterPath(path)) {
                try {
                    if (dataChangeEvent.equals(DataChangeEvent.ADD)) {
                        logger.info("master node : {} added.", path);
                        updateMasterNodes();
                    }
                    if (dataChangeEvent.equals(DataChangeEvent.REMOVE)) {
                        logger.info("master node : {} down.", path);
                        updateMasterNodes();
                        alertDao.sendServerStopedAlert(1, path, "MASTER");
                    }
                } catch (Exception ex) {
                    logger.error("MasterNodeListener capture data change and get data failed.", ex);
                }
            }
        }
    }
    /**
     * get worker node info
     *
     * @param workerNode worker node
     * @return worker node info
     */
    public String getWorkerNodeInfo(String workerNode) {
        workerNodeInfoLock.lock();
        try {
            return workerNodeInfo.getOrDefault(workerNode, null);
        } finally {
            workerNodeInfoLock.unlock();
        }
    }
}
